
#include "threadpool.h"
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <assert.h>
#include <semaphore.h>
#define DEBUG 1
#define MAX_THREAD_NUM 5

thpool * pool;

threadjob ** jobqarr;   //jobq[MAX_JOB_NUM]

jobqueuectl * jobqc;    //jobqueue control

void Thread_do_job(int threadno);

#if DEBUG
int dojobsnum[MAX_THREAD_NUM];
int givejobsnum[MAX_THREAD_NUM];
#endif

int * Process(threadjob * job,int threadno)  //workfun
{
    printf("this is %d thread do this work: %d\n",threadno,job->num);
    dojobsnum[threadno]++;
    //sleep(1);
    return 0;
}

//functions for ThreadPool
int Theadpool_create(int threadN)
{
	int vi = 0,vm = 0,threadno = 0;
	pool = NULL;	//threadpool point
	pool = (thpool *)malloc(sizeof(thpool));
	if(pool == NULL)
	{
		printf("malloc pool failed\n");
		return 1;
	}
	if(threadN <= 0)
	{
		printf("threadN is wrong\n");
		return 1;
	}
	pool->max_thread_num = threadN;
	pool->shutdown = 0;                     //0-open
	//init the job queue
	jobqarr = (threadjob**)malloc(MAX_NUM_JOB*sizeof(threadjob*));
	if(jobqarr == NULL)
	{
	    printf("job array init error\n");
	}
	memset(jobqarr,0,MAX_NUM_JOB*sizeof(threadjob*));

    //init the job queue control
    jobqc = (jobqueuectl *)malloc(sizeof(jobqueuectl));
    if(jobqc == NULL)
    {
        printf("jobqueuecontrol init error\n");
    }
    jobqc->readpos = 0;
    jobqc->writepos = 0;
    jobqc->jobcount = 0;
#if DEBUG
    for( vi = 0 ;vi < threadN ; ++vi)
    {
        dojobsnum[vi] = 0;
        givejobsnum[vi] = 0;
    }
#endif
	//init the mutex signal
	if(pthread_mutex_init(&pool->queue_lock,NULL) != 0)
	{
		printf("init pthread_mutex error %d : %s\n",errno,strerror(errno));
		return 1;
	}
	//create threads array
	pool->threadidarr = (pthread_t *)malloc(threadN*sizeof(pthread_t));
	//memset(pool->threadidarr,0,threadN * sizeof(pthread_t));
	if(pool->threadidarr == NULL)
	{
		printf("malloc threadidarr error %d : %s\n",errno,strerror(errno));
		return 1;
	}
    //init thread and the write and read signal
    pool->sem_arr = (sem_t*)malloc(threadN*sizeof(sem_t));
    assert(pool->sem_arr!=NULL);
	for(vi = 0; vi < threadN ; ++vi)
	{
	    threadno = vi;
        vm = 1;
        sem_init(&pool->sem_arr[vi],0,0);
        vm = pthread_create(&(pool->threadidarr[vi]),NULL,(void *)(&Thread_do_job),(void *)threadno);
		//vm = pthread_create(&(pool->threadidarr[vi]),NULL,(void *)(&Thread_do_job),NULL);
		printf("create %d thread\n",threadno);
		if(vm != 0)
		{
			printf("create %d thread error %d : %s\n",vi,errno,strerror(errno));
			return 1;
		}
	}
	printf("init threadpoll OK!\n");
	//sleep(5);
	return 0;
}

//add job to Queue
//arg is a struct threadjob point
int Threadpool_add_job(void *arg)
{
    int threadno = 0;
	if(arg == NULL)
	{
		printf("arg is NULL\n");
		return 1;
	}
    while(1)
    {
        if((jobqc->writepos+1)%MAX_NUM_JOB != jobqc->readpos)
            break;
    }
	pthread_mutex_lock(&pool->queue_lock);
	jobqarr[jobqc->writepos] = (threadjob *)arg;
	jobqc->writepos = (jobqc->writepos + 1)%MAX_NUM_JOB;
    jobqc->jobcount++;
    threadno = (rand()%(pool->max_thread_num));
#if DEBUG
    givejobsnum[threadno]++;
#endif
    pthread_mutex_unlock(&pool->queue_lock);
    sem_post(&pool->sem_arr[threadno]);
	return 0;
}

#if DEBUG
int PrintJobs(void)
{
    int vi = 0;
    for(vi = 0; vi < MAX_THREAD_NUM ; ++vi)
    {
        printf("%d thread get %d work and do %d work\n",vi,givejobsnum[vi],dojobsnum[vi]);
    }
    return 0;
}
#endif

int Threadpool_destroy(void)	//destroy the threadpool
{
    int vi = 0;
    if(pool->shutdown == 1)
    {
        printf("thread is already closed\n");
        return 1;
    }
    while(1)
    {
        if(jobqc->writepos == jobqc->readpos)
            break;
    }
    pthread_mutex_lock(&pool->queue_lock);        //wait for mutex lock
    pool->shutdown = 1;                           //be ready to destory threadpool
    for(vi = 0; vi < pool->max_thread_num ; ++vi)
    {
        sem_post(&pool->sem_arr[vi]);
    }
    pthread_mutex_unlock(&pool->queue_lock);
    //wait for every work thread exit
    for(vi = 0; vi< pool->max_thread_num ; ++vi)
    {
        pthread_join(pool->threadidarr[vi],NULL);
    }
    printf("begin to destroy!\n");
    for(vi = 0; vi < pool->max_thread_num ; ++vi)
    {
        sem_destroy(&pool->sem_arr[vi]);
    }
    free(jobqc);
    jobqc = NULL;    //free jobqueue control node
    free(jobqarr);
    jobqarr = NULL;  //free jobqueue
    free(pool->threadidarr);
    pool->threadidarr = NULL;   //free all work thread
    pthread_mutex_destroy(&pool->queue_lock);   //free mutex lock
    free(pool);
    pool = NULL;
    printf("free threadpool OK\n");
    return 0;
}

//thread gets job from jobqueue
void Thread_do_job(int threadno)
{
	int ret = 0;
	threadjob *thjob = NULL;
	printf("this is %d thread\n",threadno);
	if(pool == NULL)
	{
	    printf("thread pool is NULL\n");
	    exit(1);
	}

	while(1)
	{
	    sem_wait(&pool->sem_arr[threadno]);  //p
	    printf("readpos is %d\n",jobqc->readpos);
	    printf("writpos is %d\n",jobqc->writepos);
		pthread_mutex_lock(&pool->queue_lock);
		if(pool->shutdown)	//1-dead 0-alive
		{
			pthread_mutex_unlock(&pool->queue_lock);
			printf("the thread NO: %d exit\n",threadno);
			pthread_exit(&ret);
		}
		//get work
		thjob = jobqarr[jobqc->readpos];
		jobqarr[jobqc->readpos] = NULL;     //free this position from jobqueue
		jobqc->readpos = (jobqc->readpos+1)%MAX_NUM_JOB;
		jobqc->jobcount--;
		pthread_mutex_unlock(&pool->queue_lock);
		Process(thjob,threadno);      //do the job
		free(thjob);         //free the job node
        thjob = NULL;
	}
    return;
}
